/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月16日 下午8:28:27
*/
package cn.com.guage.flink.transformation;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @version:1.0.1
 * @Description: （对类进行功能描述）
 * @author: yangdechao
 * @date: datedate 2021年11月16日 下午8:28:27
 */
public class KeyByDemo {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 辽宁省,沈阳市,1000
		// 辽宁省,大连市,2000
		// 山东省,青岛市,3000
		// 山东省,烟台市,2000
		// 辽宁省,大连市,2000

		DataStreamSource<String> textStream = env.readTextFile("data/city.log");
		SingleOutputStreamOperator<ProvinceCityAndMoney2> tpDataStream = textStream.map(line -> {
			String[] fields = line.split(",");
			return ProvinceCityAndMoney2.of(fields[0], fields[1], Double.parseDouble(fields[2]));
		});
		System.out.println("------------map后结果-------------");
		// tpDataStream.print();
		System.out.println("------------map后结果-------------");
		//KeyedStream<ProvinceCityAndMoney2, Tuple> grouped = tpDataStream.keyBy("province");

//	    KeyedStream<ProvinceCityAndMoney, Tuple2<String, String>> grouped = tpDataStream.keyBy(new KeySelector<ProvinceCityAndMoney, Tuple2<String, String>>() {
//	        @Override
//	        public Tuple2<String, String> getKey(ProvinceCityAndMoney value) throws Exception {
//	            return Tuple2.of(value.getProvince(), value.getCity());
//	        }
//	    });

		KeyedStream<ProvinceCityAndMoney2, Tuple2<String, String>> grouped = tpDataStream
				.keyBy(new KeySelector<ProvinceCityAndMoney2, Tuple2<String, String>>() {
					/**
					 * 
					 */
					private static final long serialVersionUID = 7926750218774636333L;

					@Override
					public Tuple2<String, String> getKey(ProvinceCityAndMoney2 value) throws Exception {
						return Tuple2.of(value.province, value.city);
					}
				});
		// grouped.print();
		SingleOutputStreamOperator<ProvinceCityAndMoney2> summed = grouped.sum("money");

		summed.print();

		env.execute();
	}

	// 自定义的Pojo类封装数据，必须是Public
	public static class ProvinceCityAndMoney2 {

		public String province;
		public String city;
		public Double money;

		public ProvinceCityAndMoney2() {
		}

		public ProvinceCityAndMoney2(String province, String city, Double money) {
			this.province = province;
			this.city = city;
			this.money = money;
		}

		public static ProvinceCityAndMoney2 of(String province, String city, Double money) {
			return new ProvinceCityAndMoney2(province, city, money);
		}

		@Override
		public String toString() {
			return "ProvinceCityAndMoney{" + "province='" + province + '\'' + ", city='" + city + '\'' + ", money="
					+ money + '}';
		}
	}
}
